package com.dw.receiver.api;

import com.dw.common.bean.ResourceBean;
import com.dw.common.mapper.ResourceMapper;
import com.dw.common.service.KVService;
import com.dw.common.service.impl.KVServiceImpl;
import com.dw.receiver.GetAllResourcesRequest;
import com.dw.receiver.GetAllResourcesResponse;
import com.dw.receiver.ReceiverGrpc;
import com.dw.receiver.service.ReceiverService;
import com.dw.receiver.service.impl.ReceiverServiceImpl;
import com.dw.resource.GetResourceResponse;
import com.dw.resource.ResourceInfo;
import io.grpc.stub.StreamObserver;

import java.util.List;
import java.util.concurrent.BlockingDeque;

public class ReceiverGrpcImpl extends ReceiverGrpc.ReceiverImplBase {
    private final KVService kvService;
    private final ReceiverService receiverService;

    public ReceiverGrpcImpl() {
        kvService = new KVServiceImpl();
        receiverService = new ReceiverServiceImpl();
    }

    @Override
    public void getAllResources(GetAllResourcesRequest request, StreamObserver<GetAllResourcesResponse> responseObserver) {
        List<String> resources = kvService.getWithPrefix(request.getOwner());
        GetAllResourcesResponse response = GetAllResourcesResponse.newBuilder().addAllResources(resources).build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }


    @Override
    public void getResource(ResourceInfo request, StreamObserver<GetResourceResponse> responseObserver) {
        ResourceBean resource = ResourceMapper.resourceInfoToResourceBean(request);
        receiverService.getResource(resource);
        BlockingDeque<GetResourceResponse> blockingQueue = receiverService.getBlockingQueue();
        List<ResourceBean> notRequired = receiverService.getNotRequired();
        while (!notRequired.isEmpty()) {
            try {
                GetResourceResponse response = blockingQueue.take();
                responseObserver.onNext(response);
                notRequired.remove(ResourceMapper.resourceInfoToResourceBean(response.getHeader()));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        responseObserver.onCompleted();
    }
}
